package org.zhq.impl;

import org.zhq.core.IoArgs;
import org.zhq.core.IoProvider;
import org.zhq.core.Receiver;
import org.zhq.core.Sender;
import org.zhq.impl.exceptions.EmptyIoArgsException;
import org.zhq.util.CloseUtils;

import java.io.Closeable;
import java.io.IOException;
import java.nio.channels.SelectionKey;
import java.nio.channels.SocketChannel;
import java.util.concurrent.atomic.AtomicBoolean;

/**
 * SocketChannel适配器
 * <p>
 * 提供异步
 *
 * @author zhengquan
 * @date 2019/1/24
 */
public class SocketChannelAdapter implements Sender, Receiver, Closeable {

    private final AtomicBoolean closed = new AtomicBoolean(false);
    private final SocketChannel channel;
    private final IoProvider ioProvider;
    private final OnChannelStatusChangedListener listener;

    private final AbsProviderCallback inputCallback;
    private final AbsProviderCallback outputCallback;

    public SocketChannelAdapter(SocketChannel channel, IoProvider ioProvider, OnChannelStatusChangedListener listener) throws IOException {
        this.channel = channel;
        this.ioProvider = ioProvider;
        this.listener = listener;
        this.inputCallback = new InputProviderCallback(channel, SelectionKey.OP_READ, ioProvider);
        this.outputCallback = new OutputProviderCallback(channel, SelectionKey.OP_WRITE, ioProvider);
    }

    abstract class AbsProviderCallback extends IoProvider.HandleProviderCallback {

        volatile IoArgs.IoArgsEventProcessor eventProcessor;

        volatile long lastActiveTime = System.currentTimeMillis();

        AbsProviderCallback(SocketChannel channel, int ops, IoProvider ioProvider) {
            super(channel, ops, ioProvider);
        }

        @Override
        protected final boolean onProviderTo(IoArgs args) {
            if (closed.get()) {
                return false;
            }
            IoArgs.IoArgsEventProcessor processor = eventProcessor;
            if (processor == null) {
                return false;
            }
            //刷新最近活跃时间
            lastActiveTime = System.currentTimeMillis();
            args = (args == null ? processor.provideIoArgs() : args);
            try {
                if (args == null) {
                    throw new EmptyIoArgsException("提供的缓冲区为空");
                    //具体读取操作
                }
                int count = consumeIoArgs(args, channel);

                if (args.remained() && (count == 0 || args.isNeedConsumeRemaining())) {
                    attach = args;
                    return true;
                } else {
                    //写入完成回调
                    return processor.onConsumeCompleted(args);
                }
            } catch (IOException e) {
                if (processor.onConsumeFailed(e)) {
                    CloseUtils.close(SocketChannelAdapter.this);
                }
                return false;
            }
        }

        @Override
        public void fireThrowable(Throwable e) {
            final IoArgs.IoArgsEventProcessor eventProcessor = this.eventProcessor;
            if (eventProcessor == null || eventProcessor.onConsumeFailed(e)) {
                CloseUtils.close(SocketChannelAdapter.this);
            }
        }

        protected abstract int consumeIoArgs(IoArgs args, SocketChannel channel) throws IOException;
    }

    class InputProviderCallback extends AbsProviderCallback {

        InputProviderCallback(SocketChannel channel, int ops, IoProvider ioProvider) {
            super(channel, ops, ioProvider);
        }

        @Override
        protected int consumeIoArgs(IoArgs args, SocketChannel channel) throws IOException {
            return args.readFrom(channel);
        }
    }

    class OutputProviderCallback extends AbsProviderCallback {

        OutputProviderCallback(SocketChannel channel, int ops, IoProvider ioProvider) {
            super(channel, ops, ioProvider);
        }

        @Override
        protected int consumeIoArgs(IoArgs args, SocketChannel channel) throws IOException {
            return args.writeTo(channel);
        }
    }

    @Override
    public void close() throws IOException {
        if (closed.compareAndSet(false, true)) {
            //解除注册回调
            ioProvider.unregister(channel);
            //关闭
            CloseUtils.close(channel, ioProvider);
            //回调当前Channel已关闭
            listener.onChannelClosed(channel);
        }
    }


    @Override
    public void setReceiveListener(IoArgs.IoArgsEventProcessor processor) {
        inputCallback.eventProcessor = processor;
    }

    @Override
    public void postReceiveAsync() throws Exception {
        if (closed.get() || !channel.isOpen()) {
            throw new IOException("Current channel is closed!");
        }
        inputCallback.checkAttachNull();
        ioProvider.register(inputCallback);
    }

    @Override
    public long getLastReadTime() {
        return inputCallback.lastActiveTime;
    }

    @Override
    public void setSendListener(IoArgs.IoArgsEventProcessor processor) {
        outputCallback.eventProcessor = processor;
    }

    @Override
    public void postSendAsync() throws Exception {
        if (closed.get()|| !channel.isOpen()) {
            throw new IOException("Current channel is closed!");
        }
        outputCallback.checkAttachNull();
        ioProvider.register(outputCallback);
    }

    @Override
    public long getLastWriteTime() {
        return outputCallback.lastActiveTime;
    }

    public interface OnChannelStatusChangedListener {
        void onChannelClosed(SocketChannel channel) throws IOException;
    }


}
